/*
    Copyright 2009 Barnet Wagman

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package ot.s3;

import ot.*;
import java.io.*;
import java.util.*;
import org.jets3t.service.S3ServiceException;

/** An {@link Ot} that stores objects in AWS S3.
 *
 * @author Barnet Wagman
 */
public class OtS3 extends AbstractOt implements Ot, Serializable {

    static final long serialVersionUID = ConstantsOt.UID_VERSION;

    private String awsAccessID;
    private String awsSecretKey;

    public OpsS3 opsS3;
    public ManagedOps mops;

    public ContextS3 context;

    private BucketContents bucketContents;

        /** This constructor checks for the existence of the <tt>context</tt>
         *  (i.e. s3 bucket) and throws an exception if it does not exist.<p>
         *
         *  It creates a list of serializers that contains
         *  {@link StandardSerializer} and {@link GzipSerializer}.  The
         *  default serializer is set to {@link GzipSerializer}.
         *
         * @param context
         * @param awsAccessID
         * @param awsSecretKey
         * @param cacheSize
         * @param cacheDecrement
         * @throws ot.OtException
         * @throws ot.NoSuchContextException if the bucket specified in
         *     <tt>context</tt> does not exist.
         */
    public OtS3(ContextS3 context,
                String awsAccessID, String awsSecretKey,
                int cacheSize, int cacheDecrement) 
        throws OtException, NoSuchContextException {
        super(cacheSize,cacheDecrement);
        try {
            this.context = context;
            opsS3 = new OpsS3(awsAccessID, awsSecretKey);
            mops = new ManagedOps(opsS3, sert);
            readCache = new OtableCache(ConstantsOt.MAX_OBJ_IN_CACHE,
                                    ConstantsOt.CACHE_DECREMENT);
            if ( !mops.bucketExists(context.bucketName) ) {
                throw new NoSuchContextException("There is no bucket named " +
                                                 context.bucketName);
            }
            bucketContents = new BucketContents(mops);
        }
        catch(S3ServiceException xxx) { throw new OtException(xxx); }
        catch(OpFailedException xxx) { throw new OtException(xxx); }
    }

        /** This constructor checks for the existence of the <tt>context</tt>
         *  (i.e. s3 bucket) and throws an exception if it does not exist.<p>
         *
         *  It creates a list of serializers that contains
         *  {@link StandardSerializer}, {@link GzipSerializer}, and the
         *  default serializer (if it's not the standard or gzip).
         *  <tt>cacheSize=</tt>{@link ConstantsOt#MAX_OBJ_IN_CACHE},
         *  <tt>cacheDecrement=</tt>{@link ConstantsOt#CACHE_DECREMENT}.
         *
         * @param context
         * @param awsAccessID
         * @param awsSecretKey
         * @param defaultSerializer
         * @throws ot.OtException
         * @throws ot.NoSuchContextException if the bucket specified in
         *     <tt>context</tt> does not exist.
         */
    public OtS3(ContextS3 context,
                String awsAccessID, String awsSecretKey,
                Serializer defaultSerializer) 
        throws OtException, NoSuchContextException {
        super(defaultSerializer);
        try {
            this.context = context;
            opsS3 = new OpsS3(awsAccessID, awsSecretKey);
            mops = new ManagedOps(opsS3, sert);
            readCache = new OtableCache(ConstantsOt.MAX_OBJ_IN_CACHE,
                                    ConstantsOt.CACHE_DECREMENT);
            if ( !mops.bucketExists(context.bucketName) ) {
                throw new NoSuchContextException("There is no bucket named " +
                                                 context.bucketName);
            }
            bucketContents = new BucketContents(mops);
        }
        catch(S3ServiceException xxx) { throw new OtException(xxx); }
        catch(OpFailedException xxx) { throw new OtException(xxx); }
    }

         /** The constructor checks for the existence of the <tt>context</tt>
          *  (i.e. s3 bucket) and throws an exception if it does not exist.
          *
          *  It creates a list of serializers that contains
         *  {@link StandardSerializer} and {@link GzipSerializer}. The
         *  default serializer is set to {@link GzipSerializer}.
         *  <tt>cacheSize=</tt>{@link ConstantsOt#MAX_OBJ_IN_CACHE},
         *  <tt>cacheDecrement=</tt>{@link ConstantsOt#CACHE_DECREMENT}.
          *
          *  It uses {@link GzipSerializer} as its default {@link Serializer}.
          *
          * @param context
          * @param awsAccessID
          * @param awsSecretKey
          * @throws ot.OtException
          * @throws ot.NoSuchContextException
          */
    public OtS3(ContextS3 context,
                String awsAccessID, String awsSecretKey) 
        throws OtException, NoSuchContextException {
        this(context,awsAccessID,awsSecretKey,new GzipSerializer());
    }

         /** The constructor creates an instance of <tt>OtS3</tt>.  As
          *  initially created, it can only be used to call methods
          *  {@link #bucketExists(java.lang.String)},
          *  {@link #createBucket(java.lang.String)},
          *  {@link #deleteBucket(java.lang.String)}, and
          *  {@link #deleteBucket(ot.s3.ContextS3)}.  Before using any other
          *  methods, you must set the context with
          *  {@link #setContext(ot.Context)}.          
          *
          *  It creates a list of serializers that contains
         *  {@link StandardSerializer} and {@link GzipSerializer}. The
         *  default serializer is set to {@link GzipSerializer}.
         *  <tt>cacheSize=</tt>{@link ConstantsOt#MAX_OBJ_IN_CACHE},
         *  <tt>cacheDecrement=</tt>{@link ConstantsOt#CACHE_DECREMENT}.
          *
          *  It uses {@link GzipSerializer} as its default {@link Serializer}.
          *
          * @param context
          * @param awsAccessID
          * @param awsSecretKey
          * @throws ot.OtException
          * @throws ot.NoSuchContextException
          */
    public OtS3(String awsAccessID, String awsSecretKey)
        throws OtException, NoSuchContextException {
        try {
            opsS3 = new OpsS3(awsAccessID, awsSecretKey);
            mops = new ManagedOps(opsS3, sert);
            readCache = new OtableCache(ConstantsOt.MAX_OBJ_IN_CACHE,
                                    ConstantsOt.CACHE_DECREMENT);            
            bucketContents = new BucketContents(mops);
        }
        catch(S3ServiceException xxx) { throw new OtException(xxx); }       
    }

        /** Sets a
         *  <a href="http://jets3t.s3.amazonaws.com/toolkit/toolkit.html">
         *  JetS3t</a> property value. See
         *  <A href=http://jets3t.s3.amazonaws.com/toolkit/configuration.html">
         *  JetS3t Properties</a> for details.
         *
         * @param propertyName
         * @param propertyValue
         */
    public void setJetS3tProperty(String propertyName, String propertyValue)
        throws OtException {
        try {
            mops.ops.setJets3tProperty(propertyName, propertyValue);
        }
        catch(S3ServiceException xxx) { throw new OtException(xxx); }
    }

        /** {@inheritDoc}
         *
         * @param context
         * @throws ot.OtException
         * @throws ot.NoSuchContextException
         */
    public void setContext(Context context) 
        throws OtException, NoSuchContextException {
        try {
            if ( !(context instanceof ContextS3) )
                throw new OtException("OtFs requires " +
                    ContextS3.class.getName() + ", not " +
                    ((context != null)?context.getClass().getName():"null"));
            if ( !mops.bucketExists(((ContextS3) context).bucketName) ) {
                throw new NoSuchContextException("There is no bucket named " +
                                        ((ContextS3) context).bucketName);
            }
            this.context = (ContextS3) context;
        }       
        catch(OpFailedException xxx) { throw new OtException(xxx); }
    }

        /** {@inheritDoc}
         *
         * @return
         */
    public Context getContext() { return(context); }

    public boolean has(OPath path) throws OtException {
        try {
            LocO loc = new LocO(context,path);
            if ( readCache.get(loc) != null ) return(true);
            return( bucketContents.contains(loc) );
        }
        catch(Exception xxx) { throw new OtException(xxx); }
    }

        /** {@inheritDoc}<p>
              
          * @param path
          * @param obj
          * @throws ot.OtException
          */
    public void add(OPath path,Serializable obj) throws OtException {
        try {
            add(path,obj,true);
        }
        catch(ObjectExistsException ox) {}
    }

        /** {@inheritDoc}
         *
         * @param path
         * @param obj
         * @param replaceExtant
         * @throws ot.OtException
         * @throws ot.ObjectExistsException
         */
    public void add(OPath path, Serializable obj, boolean replaceExtant)
        throws OtException, ObjectExistsException {
        
        if ( !replaceExtant && has(path) ) {
            throw new ObjectExistsException(path + " already exists.");
        }
        else {
            mops.putObjToS3(new LocO(context,path), obj);
        }
    }

        /** {@inheritDoc}
         *
         * @param path
         * @return
         * @throws ot.OtException
         */
    public Serializable get(OPath path) throws OtException {
        return( get(new LocO(context,path)) );
    }

        /** {@inheritDoc}
         * 
         * @param loc
         * @return
         * @throws ot.OtException
         */
    public Serializable get(LocO loc) throws OtException {
        try {
            Serializable x = readCache.get(loc);

            if ( x != null ) return(x);
            else {
                if ( !has(loc) ) return(null);
                x = mops.getS3Obj(loc);
                if ( x != null ) {
                    readCache.add(loc, x);
                    return(x);
                }
                else return(null);
            }
        }
        catch(OpFailedException xxx) { throw new OtException(xxx); }
    }

    private boolean has(LocO loc) throws OpFailedException {
        if ( readCache.get(loc) != null ) return(true);
        return( bucketContents.contains(loc) );
    }

        /** {@inheritDoc}
         *
         * @param locs only <tt>LocO</tt>s are used to retrieve
         *  objects.  <tt>LocD</tt>s are ignored.
         * @return
         */
    public List<Serializable> get(List<Loc> locs) throws OtException {
        try {
            List<Serializable> gotten = new ArrayList<Serializable>();
            List<LocO> ungotten = new ArrayList<LocO>();

                // First try the cache
            NEXT:for( Loc lo : locs ) {
                if ( !(lo instanceof LocO) ) continue NEXT;
                Serializable x = readCache.get((LocO) lo);
                if ( x != null ) gotten.add(x);
                else ungotten.add((LocO) lo);
            }

            if ( ungotten.size() < 1 ) return(gotten);
            else {
                List<Serializable> ro = mops.getS3Objs(ungotten);
                gotten.addAll(ro);
                return(gotten);
            }         
        }
        catch(Exception xxx) { throw new OtException(xxx); }
    }

        /** This method returns
         *  a cached object or null if the object is not cached.
         *
         * @return
         */
         /**
          *
          * @param path
          * @return
          */
    public Serializable getCached(OPath path) {
        return( readCache.get(new LocO(context,path)) );
    }

        /** {@inheritDoc}
         * 
         * @param path
         * @throws ot.OtException
         */
    public void delete(OPath path) throws OtException {
        try {
            mops.deleteS3Object( context.bucketName, path.toPathString() );
        }
        catch(Exception xxx) { throw new OtException(xxx); }
    }

    public boolean bucketExists(String bucketName) throws OtException {

        try {
            return( mops.bucketExists(bucketName) );
        }
        catch(Exception x) { throw new OtException(x); }
    }

        /**
         *
         * @param bucketName
         * @return true if the bucket was created, false if it already exists.
         * @throws ot.OtException
         */
    public boolean createBucket(String bucketName) throws OtException {
        try {
            return( mops.createBucket(bucketName) );
        }
        catch(Exception x) { throw new OtException(x); }
    }

         /** Deletes an S3 bucket and <i>everything</i> in it.
          *
          * @param bucketName
          * @throws ot.OtException
          */
    public void deleteBucket(String bucketName) throws OtException {
        try {
            mops.deleteBucket(bucketName);
        }
        catch(Exception x) { throw new OtException(x); }
    }

        /** Deletes an S3 bucket and <i>everything</i> in it.
         *
         * @param context
         * @throws ot.OtException
         */
    public void deleteBucket(ContextS3 context) throws OtException {
        try {
            mops.deleteBucket(context.bucketName);
        }
        catch(Exception x) { throw new OtException(x); }
    }

        /** {@inheritDoc}
         *        
         * @return
         */
    public List<LocO> listContext() throws OtException {
        try {
            List<LocO> locs = new ArrayList<LocO>();

            List<OPath> paths = bucketContents.getList(context.bucketName);

            for ( OPath p : paths ) {
                locs.add(new LocO(context, p));
            }

            return(locs);
        }
        catch(Exception x) { throw new OtException(x); }
    }

        /** {@inheritDoc}
         *
         * @param dir the <tt>DPath</tt> of a directory.  If
         *      <tt>dir == null</tt> this method returns the
         *      <tt>Loc</tt>s of objects and directories in the top level
         *      directory in the context.
         * @return
         */
    public List<Loc> list(DPath dir) throws OtException {
        try {
            List<OPath> opb = bucketContents.getList(context.bucketName);

            if ( dir == null ) return(listTopLevel(opb));

            String dirString = dir.toPathString();

            HashMap<String,Loc> ht = new HashMap<String,Loc>();
            int dirLen = dir.pathnames.length;
            for ( OPath p : opb ) {
                if ( p.toPathString().startsWith(dirString) ) {
                    if ( p.pathnames.length == (dirLen + 1) ) {
                                    // It's an object in dir
                        ht.put(p.toPathString(),new LocO(context,p));
                    }
                    else { // It's a dir                   
                        String[] child = new String [dirLen + 1];
                        System.arraycopy(p.pathnames,0,child,0,dirLen + 1);
                        LocD d = new LocD(context,new DPath(child));
                        ht.put(d.getPath().toPathString(),d);
                    }
                }
            }

            List<Loc> locs = new ArrayList<Loc>();
            for ( Iterator<Loc> it = ht.values().iterator(); it.hasNext(); ) {
                locs.add(it.next());
            }

            return(locs);
        }
        catch(Exception xxx) { throw new OtException(xxx); }
    }

    private List<Loc> listTopLevel(List<OPath> oPaths) {

        List<Loc> locs = new ArrayList<Loc>();
        HashMap<String,String> dirHt = new HashMap<String,String>();
        
        for ( OPath p : oPaths ) {
            if ( p.pathnames.length == 1 ) { // It's a top level object
                locs.add(new LocO(context,p));
            }
            else {
                dirHt.put(p.pathnames[0],p.pathnames[0]);
            }
        }

        for ( Iterator<String> it = dirHt.keySet().iterator(); it.hasNext(); ) {
            locs.add(new LocD(context,new DPath(new String[] {it.next()})));
        }

        return(locs);
    }

        /** {@inheritDoc}
         *
         * @param dir the <tt>DPath</tt> of a directory.  If
         *      <tt>dir == null</tt> this method returns the
         *      <tt>Loc</tt>s of objects and directories in the top level
         *      directory in the context.
         * @return
         */
    public List<Loc> listTree(DPath dir) throws OtException {

        try {
            List<OPath> opb = bucketContents.getList(context.bucketName);

                // First get objects' paths.
            List<OPath> oPaths = null;
            if ( dir != null ) oPaths = listOPathsInTree(dir);
            else oPaths = bucketContents.getList(context.bucketName);

                // Get the dirs
            HashMap<String,DPath> dirHt = new HashMap<String, DPath>();

            for ( OPath p : oPaths ) {
                String[] dirPathNames = new String[p.pathnames.length - 1];
                System.arraycopy(p.pathnames,0,dirPathNames,0,dirPathNames.length);
                DPath d = new DPath(dirPathNames);
                dirHt.put(d.toPathString(),d);
            }

            List<Path> paths = new ArrayList<Path>();
            for ( OPath o : oPaths ) paths.add(o);

            for ( Iterator<DPath> it = dirHt.values().iterator(); it.hasNext(); ) {
                paths.add(it.next());
            }

            Collections.sort(paths);

            List<Loc> locs = new ArrayList<Loc>();
            for( Path p : paths ) {
                if ( p instanceof OPath ) locs.add(new LocO(context, p));
                else if ( p instanceof DPath ) locs.add(new LocD(context, p));
                else throw new RuntimeException("Impossible case: path is " +
                    ((p != null)?p.getClass().getName():"null"));                        
            }

            return(locs);
        }
        catch(Exception xxx) { throw new OtException(xxx); }
    }

    public BucketContents getBucketContents() { return(bucketContents); }

    private List<OPath> listOPathsInTree(DPath dir) throws Exception {

        List<String> s3Keys = mops.listS3Objects(new LocD(context, dir));

        List<OPath> oPaths = new ArrayList<OPath>();

        for ( String k : s3Keys ) {
            OPath p = new OPath(k.split("/"));
            oPaths.add(p);
        }
        return(oPaths);
    }

    private boolean isObjInDir(DPath dir, OPath p) {

        if ( p.pathnames.length != (dir.pathnames.length + 1) )
            return(false);

        return( p.toPathString().startsWith(dir.toPathString()) );
    }

    private DPath getDirInDir(DPath dir, OPath p) {

        if ( !p.toPathString().startsWith(dir.toPathString()) )
            return(null);

        String[] ds = new String [dir.pathnames.length + 1];
        System.arraycopy(p.pathnames,0,ds,0,dir.pathnames.length + 1);
        return( new DPath(ds) );
    }
}
